package com.shujia.core

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo02WordCountSubmit extends Logging {
  def main(args: Array[String]): Unit = {
    // 1、构建Spark环境
    // 配置Spark任务
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo01WordCount") // 设置任务名
    //    conf.setMaster("local") // 设置Spark的运行方式
    // 创建SparkContext --> Spark程序的入口
    val sc: SparkContext = new SparkContext(conf)

    // 2、构建Spark程序
    // 读取文件
    /**
     * 如果是Local本地模式：该路径为本地的路径，需要在我的电脑里配置系统环境变量HADOOP_HOME
     * 如果是以集群模式运行：该路径为HDFS的路径
     * 不管是哪种方式运行，都是使用Hadoop的那一套读取文件的方式（切片、格式化数据）
     */
    /**
     * RDD：弹性分布式数据集，它不真正存储数据，它只是一个编程模型（统一了所有的方法，所有的Spark的程序都是基于RDD进行处理的）
     * 在使用层面，可以把它当成Scala中的List去使用，List有的常用的方法，RDD都有
     * 例如都有：map、flatMap、filter、foreach、sortBy、groupBy
     * 这些方法在Spark中统称为 算子
     * 算子可以分为两类：转换算子、行为算子
     * 算子内部的代码都会被封装成一个个的Task最终在Executor中执行
     * 算子外部的代码都在Driver端执行
     */
    val lineRDD: RDD[String] = sc.textFile("/data/spark/wc/input")

    // 将每个单词提取出来
    val wordsRDD: RDD[String] = lineRDD.flatMap(_.split(","))

    // 按每个单词进行分组
    val groupRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)


    // 统计每个单词的数量
    val wordCntRDD: RDD[String] = groupRDD.map(kv => s"${kv._1},${kv._2.size}")

    // 将结果输出
    wordCntRDD.foreach(println)
    wordCntRDD.foreach(line => {
      log.info(line)
    })

    // 判断路径是否存在 存在即删除
    val hadoopConf: Configuration = new Configuration()
    hadoopConf.set("fs.defaultFS", "hdfs://master:9000")
    val fs: FileSystem = FileSystem.get(hadoopConf)
    val outPath: Path = new Path("/data/spark/wc/output")
    if (fs.exists(outPath)) {
      fs.delete(outPath, true)
    }

    // 一般计算的最终结果不会直接打印，通常是会将结果保存到HDFS的某个路径
    wordCntRDD.saveAsTextFile("/data/spark/wc/output")

    println("这是在Driver执行的代码")
    log.info("这是通过Spark提供的log方法打印的日志")


  }

  /**
   * 将任务提交至集群：
   * 1、将这行代码conf.setMaster("local")注释
   * 2、将本地路径换成HDFS的路径（需要先在HDFS上准备数据）
   * 3、将代码通过maven打成jar包
   * 4、上传jar包并提交任务
   * yarn-client模式：spark-submit --class com.shujia.core.Demo02WordCountSubmit --master yarn-client --executor-memory 512M --num-executors 2 Spark-1.0.jar
   * yarn-cluster模式：spark-submit --class com.shujia.core.Demo02WordCountSubmit --master yarn-cluster --executor-memory 512M --num-executors 2 Spark-1.0.jar
   *
   */

}
